/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.nhncorp.neptune.changelog.pipe;

import java.io.IOException;
import java.lang.reflect.UndeclaredThrowableException;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.List;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.nhncorp.neptune.changelog.ChangeLogServer;
import com.nhncorp.neptune.common.testhelper.FaultInjectionProxy;
import com.nhncorp.neptune.common.testhelper.ProxyExceptionHelper;

public class Bulk implements BulkTestIF {
  public static final int DEFAULT_BUFFER_SIZE = 256 * 1024; // 256KB
  static final Log LOG = LogFactory.getLog(Bulk.class);

  static enum OperationResult {
    completed, partially
  }

  List<ByteBuffer> bufferList = new ArrayList<ByteBuffer>(10);
  int currentReadBufIndex = 0;
  int currentWriteBufIndex = 0;
  int writtenPos = 0;
  byte[] txId = null;
  int seq = -1;
  int totalDataSize = -1;
  boolean readingHeaderDone = false;
  int numRead = 0;

  BulkTestIF testProxy;
  
  public Bulk() {
    testProxy = this;
    if (ChangeLogServer.IS_TEST_MODE) {
      testProxy = FaultInjectionProxy.wrap(this, BulkTestIF.class);
    }
    addNewBuffersToList();
  }

  public OperationResult read(SocketChannel ch) throws IOException {
    if (bufferList.isEmpty()) {
      throw new IOException("Pipe is closed");
    }

    int curNumRead = 0;

    while (true) {
      ByteBuffer buf = bufferList.get(currentReadBufIndex);

      try {
        curNumRead = testProxy.readFromChannel(ch, buf);
      } catch(UndeclaredThrowableException e) {
        ProxyExceptionHelper.handleException(e, LOG);
      }

      numRead += curNumRead;

      if (!readingHeaderDone) {
        readingHeaderDone = readHeader(buf);
      }

//      LOG.debug("numRead : " + numRead);
      if (buf.hasRemaining()) {
        return totalDataSize >= 0 && numRead >= totalDataSize ? OperationResult.completed
            : OperationResult.partially;
      }

      currentReadBufIndex++;
//      LOG.debug("buffer change, index[" + currentReadBufIndex + "]");

      if (bufferList.size() <= currentReadBufIndex) {
        addNewBuffersToList();
//        LOG.debug("append new buffer. bufferList.size : " + bufferList.size());
      }
    }
  }

  // by sangchul test때문에 visibility를 public으로 변경
  public int readFromChannel(SocketChannel ch, ByteBuffer buf) throws IOException {
    int curNumRead;
    if ((curNumRead = ch.read(buf)) < 0) {
      throw new IOException("End of stream");
    }
    return curNumRead;
  }

  private boolean readHeader(ByteBuffer buf) {
    ByteBuffer headerBuf = buf.duplicate();
    headerBuf.flip();

    if (seq < 0 && txId == null && 8 <= headerBuf.limit()) {
      seq = headerBuf.getInt();
//      LOG.debug("seq#" + seq);
      txId = new byte[headerBuf.getInt()];
    }

    if (txId != null && txId.length + 8 <= headerBuf.limit()) {
      headerBuf.position(8);
      headerBuf.get(txId);
//      LOG.debug("tx id : " + new String(txId).trim());
    }

    if (txId != null && totalDataSize < 0 && txId.length + 8 + 4 <= headerBuf.limit()) {
      headerBuf.position(8 + txId.length);
      totalDataSize = headerBuf.getInt();
//      LOG.debug("total data size : " + totalDataSize);

      return true;
    }

    return false;
  }

  boolean printNumWritten = false;
  int totalNumWritten = 0;

  public OperationResult write(SocketChannel ch) throws IOException {
    if (bufferList.size() == 0) {
      throw new IOException("Pipe is closed");
    }

    int numWritten = 0;

    while (true) {
      ByteBuffer readBuf = bufferList.get(currentWriteBufIndex);
      ByteBuffer writeBuffer = readBuf.duplicate();
      writeBuffer.position(writtenPos);
      writeBuffer.limit(readBuf.position());

      numWritten = ch.write(writeBuffer);

      writtenPos += numWritten;
      totalNumWritten += numWritten;
//      LOG.debug("current write buffer index : " + currentWriteBufIndex + ", totalNumWritten : " + totalNumWritten);

      if (writeBuffer.hasRemaining() == false) {
        if (writeBuffer.limit() == writeBuffer.capacity()
            && currentWriteBufIndex < currentReadBufIndex) {
          currentWriteBufIndex++;
          writtenPos = 0;
//          LOG.debug("write buffer is changed. currentWriteBufIndex : " + currentWriteBufIndex);
        } else {
//          LOG.debug("OperationResult.completed, writeBuffer.limit() : " + writeBuffer.limit() + ", writeBuffer.position() : " + writeBuffer.position());
          return OperationResult.completed;
        }
      } else {
//        LOG.debug("OperationResult.partially, writeBuffer.limit() : " + writeBuffer.limit() + ", writeBuffer.position() : " + writeBuffer.position());
        return OperationResult.partially;
      }
    }
  }

  // by sangchul
  // 이 메소드는 pipe를 다루는 쓰레드와 다른 쓰레드가 들어온다.
  // 하지만, 그 다른 쓰레드는 pipe에 이미 정해진 수의 바이트가 들어오기 전까진 block하고 있다가
  // 그 후에 이 메소드를 호출하기 때문에 bufferList가 여러 쓰레드에 의해 공유되는 상황은 발생하지 않는다.
  public ByteBuffer[] getBufferArray() {
    ByteBuffer[] bufferArray = bufferList.toArray(new ByteBuffer[0]);

    for (int i = 0; i < bufferArray.length; i++) {
      bufferArray[i].flip();
    }

    return bufferArray;
  }

  public void clear() {
    ByteBuffer buf = bufferList.get(0);
    buf.clear();
    
    if (bufferList.size() > 1) {
      List<ByteBuffer> returnedList = bufferList.subList(1, bufferList.size() - 1);
      BufferPool.singleton().returnBuffer(returnedList.toArray(new ByteBuffer[0]));
      returnedList.clear();
    }

    bufferList.clear();
    bufferList.add(buf);
    
    currentReadBufIndex = 0;
    currentWriteBufIndex = 0;
    writtenPos = 0;
    readingHeaderDone = false;
    totalDataSize = -1;
    numRead = 0;
    txId = null;
    seq = -1;
    totalNumWritten = 0;
  }

  private void returnBuffer() {
    ByteBuffer[] bufferArray = bufferList.toArray(new ByteBuffer[0]);
    bufferList.clear();
    BufferPool.singleton().returnBuffer(bufferArray);
  }

  private void addNewBuffersToList() {
    for (ByteBuffer buf : BufferPool.singleton().getBuffer(DEFAULT_BUFFER_SIZE)) {
      bufferList.add(buf);
    }
  }

  public void close() {
    returnBuffer();
    bufferList = null;
  }

  public boolean isWrittingDone() {
    return totalDataSize == totalNumWritten;
  }
}
